//
// Created by 28365 on 2025/3/10.
//

#include <iostream>
#include "../include/thread/ThreadPool.h"
#include "../include/thread/DefaultThreadFactory.h"

ThreadPool::ThreadPool(size_t _coreSize, size_t _queueSize){
    coreSize = _coreSize;
    queueSize = _queueSize;
    threadFactory = new DefaultThreadFactory;
}

ThreadPool::~ThreadPool() {
    {
        unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    await();

    delete threadFactory;
}

void ThreadPool::execute(function<void()> &&runnable) {

    {
        unique_lock<std::mutex> lock(queue_mutex);
        if (tasks.size() >= queueSize){
            queue_condition.wait(lock,[this]{return stop || tasks.size() < queueSize;});
        }
        if (stop) {
            throw runtime_error("enqueue on stopped ThreadPool");
        }

        if (workers.size() < coreSize){
            //当核心线程未满时启动消费者线程
            startTask();
        }

        tasks.emplace(runnable);
    }
    condition.notify_one();
}

void ThreadPool::startTask() {
    thread *t = threadFactory->newThread([this] {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(this->queue_mutex);

                if (this->tasks.empty() && !this->condition.wait_for(lock,chrono::minutes(1),[this] { return this->stop || !this->tasks.empty();})){
                    return;
                }

                if (workers.size() > coreSize){
                    for (auto it = workers.begin(); it != workers.end();it++) {
                        if (const_cast<thread*>(*it.base())->get_id() == this_thread::get_id()){
                            workers.erase(it);
                            return;
                        }
                    }
                }

                //停止标签且任务队列为空结束线程
                if (this->stop && this->tasks.empty()) {
                    return;
                }

                task = std::move(this->tasks.front());
                this->tasks.pop();
            }
            task();
            //唤醒由于满队列而阻塞的函数
            queue_condition.notify_one();
        }
    });
    workers.push_back(t);
}

void ThreadPool::shutdown() {
    this->stop = true;
}

void ThreadPool::await() {
    if (!stop){
        return;
    }
    condition.notify_all();
    for (thread *worker : workers) {
        if (worker->joinable()){
            worker->join();
        }
        delete worker;
    }
    workers.clear();
}

size_t ThreadPool::getWorkerSize() {
    return workers.size();
}

size_t ThreadPool::getCoreSize() const {
    return coreSize;
}

void ThreadPool::setCoreSize(size_t _coreSize) {
    {
        std::unique_lock<std::mutex> lock(this->queue_mutex);
        ThreadPool::coreSize = _coreSize;
    }
}

void ThreadPool::shutdownAndDiscard() {
    shutdown();
    {
        std::unique_lock<std::mutex> lock(this->queue_mutex);
        queue<function<void()>> emptyQueue;
        swap(tasks,emptyQueue);
    }
}
